package com.ds.mq.consumer.config;


import com.ds.mq.consumer.converter.CustomConverter;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.ssl.SslBundles;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.stereotype.Component;

@Component
@RequiredArgsConstructor
public class KafkaConfig {
    private final KafkaProperties properties;
    private final SslBundles sslBundles;

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> factory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 转换成list批量消费
        factory.setBatchListener(properties.getListener().getType() == KafkaProperties.Listener.Type.BATCH);
        // 自定义转换器
        factory.setBatchMessageConverter(new CustomConverter());
        // 消费者线程数
        factory.setConcurrency(properties.getListener().getConcurrency());
        // 自动提交
        factory.getContainerProperties().setAckMode(properties.getListener().getAckMode());
        return factory;
    }

    //@Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties(sslBundles));
    }
}
